package com.sui.bigdata.rtcadmin.util;

import com.feidee.fdcommon.util.ToolUtils;
import com.sui.bigdata.flink.table.client.JobClient;
import com.sui.bigdata.rtcadmin.async.JobAlarm;
import com.sui.bigdata.rtcadmin.dto.JobConfigDto;
import com.sui.bigdata.rtcadmin.repository.model.JobConfig;
import com.sui.bigdata.rtcadmin.timing.Crontab;
import org.apache.flink.api.common.JobStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

import static com.sui.bigdata.rtcadmin.util.SendMsgUtils.*;

/**
 * @author YongChen
 * @date 2019/12/13 9:50
 * @description
 * @email yong_chen@sui.com
 */

@Component
public class AlarmUtils {

    private static Logger logger = LoggerFactory.getLogger(AlarmUtils.class);

    private static String FLINK_CONF = "/conf";

    private static String DISCONNECT = "DISCONNECT";

    @Autowired
    private JobActionUtils jobActionUtils;

    @Value("${alarm.msg.admin.url}")
    private String alarmMsgAdminUrl;

    @Autowired
    private RedisUtils redisUtils;

    public static Boolean exist(String var, String... arguments) {
        for (String argument : arguments) {
            if (var.equals(argument)) {
                return Boolean.TRUE;
            }
        }
        return Boolean.FALSE;
    }

    public static Boolean doMonitor(JobConfig jobConfig) {
        if (ToolUtils.isNotNull(jobConfig.getAppId())
                && exist(jobConfig.getStatus(), "RUNNING")) {
            return true;
        }
        return false;
    }

    public static Boolean doUpdate(JobConfig jobConfig) {
        if (ToolUtils.isNotNull(jobConfig.getAppId())
                && exist(jobConfig.getStatus(), "RUNNING", "ACCEPT", "RESTARTING", "DISCONNECT",
                "KILLED","FAILED", "INITIALIZING")) {
            return true;
        }
        return false;
    }

    public void getJobStatus(List<JobConfig> jobConfigs,
                             String flinkIntallPath,
                             String yarnConfPath,
                             JobAlarm jobAlarm)
            throws Exception {

        for (JobConfig jobConfig : jobConfigs) {
            String appId = jobConfig.getAppId();
            String jobId = jobConfig.getJobId();
            String jobName = jobConfig.getJobName();
            JobStatus jobStatus = null;

            YarnApplicationState yarnApplicationState = null;
            List<Object> statusList = null;
            try {
                statusList = JobClient.getYarnStatus(appId, yarnConfPath);
                yarnApplicationState = (YarnApplicationState)statusList.get(0);
            } catch (ApplicationNotFoundException e1) {
                yarnApplicationState = YarnApplicationState.KILLED;
                logger.error(jobName + "：" + appId + " not found exception .");
            } catch (YarnException e2) {
                jobConfig.setStatus(DISCONNECT);
                if (redisUtils.setScheduler("yarnDisconnect", "yarnDisconnect", 24 * 60 * 60L)) {
                    sendYarnDisconnectMsg(alarmMsgAdminUrl);
                }
                logger.error(" yarn disconnect: ", e2);
                continue;
            }

            switch (yarnApplicationState) {
                case KILLED:
                    jobConfig.setStatus(YarnApplicationState.KILLED.toString());
                    Crontab.threadPool.execute(JobAlarm.build(jobAlarm, jobConfig, yarnApplicationState.toString()));
                    logger.info(jobName + " has been killed");
                    break;
                case FAILED:
                    jobConfig.setStatus(YarnApplicationState.FAILED.toString());
                    Crontab.threadPool.execute(JobAlarm.build(jobAlarm, jobConfig, yarnApplicationState.toString()));
                    break;
                case FINISHED:
                    FinalApplicationStatus finalStatus = (FinalApplicationStatus) statusList.get(1);
                    jobConfig.setStatus(finalStatus.toString());
                    if(FinalApplicationStatus.FAILED.equals(finalStatus) || FinalApplicationStatus.KILLED.equals(finalStatus)){
                        Crontab.threadPool.execute(JobAlarm.build(jobAlarm, jobConfig, finalStatus.toString()));
                    }
                    break;
                default:
                    try {
                        jobStatus = JobClient.getFlinkStatus(jobId, appId, flinkIntallPath, yarnConfPath);
                    } catch (Exception e) {
                        logger.error("flink状态获取失败！将状态改为：FAILED，jobName：{}，jobId:{},appId:{}",
                                jobName, jobId, appId, e);
                        jobStatus = JobStatus.FAILED;
                    }
                    jobConfig.setStatus(jobStatus.toString());
                    break;
            }
            if ((yarnApplicationState == YarnApplicationState.RUNNING) &&
                    (jobStatus == JobStatus.FAILED ||
                            jobStatus == JobStatus.CANCELED)) {
                JobClient.killApplication(appId, yarnConfPath);
                Crontab.threadPool.execute(JobAlarm.build(jobAlarm, jobConfig, jobStatus.toString()));
                logger.info(jobName + " not running ,so kill ");
            }
            if ((yarnApplicationState == YarnApplicationState.RUNNING) &&
                    (jobStatus == JobStatus.CANCELLING ||
                            jobStatus == JobStatus.FAILING)) {
                jobConfig.setStatus(JobStatus.RUNNING.toString());
            }
        }
    }

    public void restartJob(String jobName, String alarmUrl, JobConfig jobConfig, String webUrl, String jobStatus, int todayRestartTimes) {
        JobConfigDto jobConfigDto = new JobConfigDto();
        try {
            jobActionUtils.restart(jobConfigDto.setJobName(jobName));
            sendJobRestartSuccessMsg(alarmUrl, jobName, jobStatus, jobConfig.getAppId(), todayRestartTimes);
        } catch (Exception e) {
            sendJobRestartFailMsg(alarmUrl, jobName, jobConfig.getDeveloper(), jobStatus, webUrl, todayRestartTimes);
            logger.error(" {} change failed to restart fail.", jobName, e);
        }
    }
}
